#pragma once
#include "../common/mq_msg/mq_msg.pb.h"
#include "../common/helper.hpp"
#include "../common/log.hpp"

#include <iostream>
#include <list>
#include <sstream>
#include <mutex>

namespace xzt
{
    using MessagePtr = std::shared_ptr<xzt::Message>;
    using MessageMap = std::map<std::string, MessagePtr>;
// using std::size_t;
#define DATAFILE_SUFFIX ".mqd"
#define TMPFILE_SUFFIX ".mqd.tmp"
#define MSG_VALID_FLAG "1"
#define MSG_INVALID_FLAG "0"

    class MessageMapper
    {
    public:
        MessageMapper(const std::string &basedir, const std::string &queue_name) : _queue_name(queue_name)
        {
            std::stringstream ss;
            ss << basedir;
            if (basedir.back() != '/')
            {
                ss << "/";
            }
            ss << queue_name;
            _datafile = ss.str() + DATAFILE_SUFFIX;
            _tmpfile = ss.str() + TMPFILE_SUFFIX;

            if (!xzt::FileHelper(_datafile).exist())
            {
                // 创建目录
                if (!xzt::FileHelper::createDirectory(basedir))
                {
                    ELOG("create directory %s failed", basedir.c_str());
                    exit(EIO); // 目录不存在
                }
                // 创建消息文件
                if (!createMsgFile())
                {
                    ELOG("create message file %s failed", _datafile.c_str());
                    exit(EIO); // 其他错误
                }
            }
        }

        bool createMsgFile()
        {
            if (xzt::FileHelper(_datafile).exist()) // 防止文件已存在，直接创建会覆盖原文件
            {
                return true;
            }
            if (!xzt::FileHelper::createFile(_datafile))
            {
                ELOG("MassageMapper::createMsgFile() -> 创建队列数据文件 %s 失败", _datafile.c_str());
                return false;
            }
            return true;
        }

        void removeMsgFile()
        {
            if (!xzt::FileHelper::removeFile(_datafile))
            {
                ELOG("MassageMapper::removeMsgFile() -> 删除队列数据文件 %s 失败", _datafile.c_str());
            }
            if (!xzt::FileHelper::removeFile(_tmpfile))
            {
                ELOG("MassageMapper::removeMsgFile() -> 删除队列临时文件 %s 失败", _tmpfile.c_str());
            }
        }

        bool insert(MessagePtr &msg)
        {
            return _insert(_datafile, msg);
        }
        bool remove(MessagePtr &msg)
        {
            // 1.将msg的valid标志位设置为MSG_INVALID_FLAG
            msg->mutable_payload()->set_valid(MSG_INVALID_FLAG);
            // 2.对msg进行序列化，获取序列化后的字符串
            std::string serialized_msg = msg->payload().SerializeAsString();
            if (serialized_msg.size() != msg->length())
            {
                ELOG("MassageMapper::remove() -> 消息长度不匹配，可能覆盖其他消息，无法删除");
                return false;
            }
            // 3.用序列化后的字符串替换掉原来的消息（覆盖），替换了标志位
            FileHelper file_helper(_datafile);
            if (!file_helper.write(serialized_msg.c_str(), msg->offset(), serialized_msg.size()))
            {
                ELOG("MassageMapper::remove() -> 写入队列数据文件 %s 失败", _datafile.c_str());
                return false; // 写入失败
            }
            return true;
        }

        // 垃圾回收
        std::list<MessagePtr> gc()
        {
            // 1.加载文件中所有有效消息 存储格式：4字节长度|数据|4字节长度|数据|...
            std::list<MessagePtr> gc_list;
            bool ret = _loadAllMsg(gc_list, _datafile);
            DLOG("gc_list size: %ld", gc_list.size());

            // 2.将有效数据进行序列化，写入临时文件
            FileHelper tmp_file_helper(_tmpfile);
            FileHelper::createFile(_tmpfile);
            for (auto &p_msg : gc_list)
            {
                // _insert()会将MessagePtr转化为序列化后的字符串，并写入文件末尾
                _insert(_tmpfile, p_msg);
            }

            // 3.删除原文件
            if (!FileHelper::removeFile(_datafile))
            {
                ELOG("MassageMapper::gc() -> 删除队列数据文件 %s 失败", _datafile.c_str());
            }
            // 4.重命名临时文件为原文件名
            if (!tmp_file_helper.rename(_datafile))
            {
                ELOG("MassageMapper::gc() -> 重命名队列数据文件 %s 失败", _tmpfile.c_str());
            }

            return gc_list;
        }

    private:
        // 加载文件中所有有效消息 存储格式：8字节长度|数据|8字节长度|数据|...
        bool _loadAllMsg(std::list<MessagePtr>& msg_list, const std::string &filename)
        {
            bool ret = false;

            FileHelper file_helper(filename);
            size_t file_size = file_helper.size();
            size_t offset = 0;
            while (offset < file_size)
            {
                // <1>读取消息长度
                size_t msg_len = 0;
                if (!file_helper.read((char *)&msg_len, offset, sizeof(msg_len)))
                {
                    ELOG("MassageMapper::_loadAllMsg() -> 读取队列数据文件 %s 失败", filename.c_str());
                    break; // 读取失败
                }
                offset += sizeof(msg_len);
                // <2>读取消息内容
                std::string msg_str(msg_len, '\0');

                if (!file_helper.read(&msg_str[0], offset, msg_len))
                {
                    ELOG("MassageMapper::_loadAllMsg() -> 读取队列数据文件 %s 失败", filename.c_str());
                    break; // 读取失败
                }
                offset += msg_len;
                // <3>反序列化消息
                xzt::MessagePtr p_msg = std::make_shared<xzt::Message>();
                if (!p_msg->mutable_payload()->ParseFromString(msg_str))
                {
                    ELOG("MassageMapper::_loadAllMsg() -> 反序列化消息失败");
                    break; // 反序列化失败
                }

                // <4>判断消息是否有效
                if (p_msg->payload().valid() == MSG_VALID_FLAG)
                {
                    // 有效消息，放入msg_list
                    msg_list.push_back(p_msg);
                }
                ret = true;
            }
            return ret;
        }

        // 将消息插入到文件中  存储格式：8字节长度|数据|8字节长度|数据|...
        bool _insert(const std::string &filename, MessagePtr &msg)
        {
            // 新增数据到文件直接追加到文件末尾
            // 1.对消息序列化，获取序列化后的字符串
            std::string serialized_msg = msg->payload().SerializeAsString();
            // 2.获取文件长度
            FileHelper file_helper(filename);
            size_t file_size = file_helper.size();
            size_t msg_size = serialized_msg.size();
            // 写入文件
            // <1>写入消息长度
            if (!file_helper.write((char *)&msg_size, file_size, sizeof(msg_size)))
            {
                ELOG("MassageMapper::insert() -> [消息长度]写入队列数据文件 %s 失败", filename.c_str());
                return false; // 写入失败
            }
            // <2>写入消息内容
            file_size += sizeof(msg_size);
            if (!file_helper.write(serialized_msg.c_str(), file_size, serialized_msg.size()))
            {
                ELOG("MassageMapper::insert() -> [消息内容]写入队列数据文件 %s 失败", filename.c_str());
                return false; // 写入失败
            }

            // 4.更新message对象中实际的offset和length
            msg->set_offset(file_size);
            msg->set_length(serialized_msg.size());
            return true;
        }

    private:
        std::string _queue_name;
        std::string _datafile;
        std::string _tmpfile;
    };

    // 以队列为单位，管理消息
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;

    public:
        QueueMessage(const std::string &basedir, const std::string &queue_name) : _queue_name(queue_name), _valid_count(0), _total_count(0), _mapper(basedir, queue_name) {}

        void recovery()
        {
            // 加载持久化消息
            _pendingToPush_list = _mapper.gc(); // 对持久化文件进行垃圾回收
            for (auto &p_msg : _pendingToPush_list)
            {
                // 持久化消息放入durable_msg_map
                _durable_msg_map.insert(std::make_pair(p_msg->payload().propeties().id(), p_msg));
            }
            _valid_count = _total_count = _pendingToPush_list.size();
            ILOG("QueueMessage::recovery() -> 加载持久化消息成功，共 %ld 条", _total_count);
        }

        // 参数对应：消息属性、消息体
        bool insert(const xzt::BasicPropeties *basic_props, const std::string &body, DeliveryMode delivery_mode = DeliveryMode::DURABLE)
        {
            // 1.创建消息对象
            MessagePtr p_msg = std::make_shared<xzt::Message>();
            //<1>设定消息载荷
            p_msg->mutable_payload()->set_body(body);
            //<2>设定消息属性
            if (basic_props != nullptr)
            {
                p_msg->mutable_payload()->mutable_propeties()->set_id(basic_props->id());
                p_msg->mutable_payload()->mutable_propeties()->set_routing_key(basic_props->routing_key());
                p_msg->mutable_payload()->mutable_propeties()->set_delivery_mode(basic_props->delivery_mode());
            }
            else
            {
                p_msg->mutable_payload()->mutable_propeties()->set_id(UUIDHelper::uuid());
                p_msg->mutable_payload()->mutable_propeties()->set_routing_key("");
                p_msg->mutable_payload()->mutable_propeties()->set_delivery_mode(delivery_mode);
            }
            // 2.判断消息是否需要持久化存储
            std::lock_guard<std::mutex> lock(_mutex);
            if (p_msg->payload().propeties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 设置Message对象的valid标志位为MSG_VALID_FLAG
                p_msg->mutable_payload()->set_valid(MSG_VALID_FLAG);
                // 持久化存储消息
                if (!_mapper.insert(p_msg))
                {
                    ELOG("QueueMessage::insert() -> 持久化存储消息失败");
                    return false; // 持久化存储失败
                }
                _durable_msg_map.insert(std::make_pair(p_msg->payload().propeties().id(), p_msg));

                // 文件中消息数量加1
                _total_count++;
                // 有效消息数量加1
                _valid_count++;
            }
            // 3.内存管理
            _pendingToPush_list.push_back(p_msg);
            return true;
        }

        // 确认（删除）消息 -- 从待确认队列中删除消息，并判断是否需要回收
        bool remove(const std::string &msg_id) // 每次删除消息后，判断是否需要回收
        {
            std::lock_guard<std::mutex> lock(_mutex);
            // 1.从待确认队列中查找消息
            auto it = _waitack_msg_map.find(msg_id);
            if (it == _waitack_msg_map.end())
            {
                ILOG("QueueMessage::remove() -> 消息 %s 不存在于待确认队列中", msg_id.c_str());
                return true; // 消息不存在
            }

            // 2.判断消息是否需要持久化存储，如果需要，则从持久化存储中删除
            MessagePtr p_msg = it->second;
            if (p_msg->payload().propeties().delivery_mode() == DeliveryMode::DURABLE)
            {
                if (!_mapper.remove(p_msg))     // 文件中修改消息的valid标志位为MSG_INVALID_FLAG，并覆盖原消息
                {
                    ELOG("QueueMessage::remove() -> 删除持久化消息失败");
                    return false; // 删除持久化消息失败
                }
                _durable_msg_map.erase(msg_id);
                _valid_count--; // front获取时不需要对持久化的有效消息数量减1，删除（确认）消息才更改

                // 垃圾回收
                gc(); // 内部会判断是否需要进行垃圾回收
            }
            // 3.从待确认队列中删除消息
            _waitack_msg_map.erase(it);

            return true;
        }

        // 获取队首消息
        MessagePtr front()  // front()不会对持久化的有效消息数量减1，删除（确认）消息才更改
        {
            std::lock_guard<std::mutex> lock(_mutex);
            // 1.获取队首消息
            MessagePtr p_msg = nullptr;
            if (!_pendingToPush_list.empty())
            {
                p_msg = _pendingToPush_list.front();
            }
            // 2.将对象从待推送列表中移除，_waitcak_msg_map中添加
            if (p_msg != nullptr)
            {
                _pendingToPush_list.pop_front();

                _waitack_msg_map.insert(std::make_pair(p_msg->payload().propeties().id(), p_msg));
            }

            return p_msg;
        }

        // 获取待推送消息数量
        size_t getable_count()
        {
            std::lock_guard<std::mutex> lock(_mutex);
            return _pendingToPush_list.size();
        }

        // 获取总消息数量
        size_t total_count()
        {
            std::lock_guard<std::mutex> lock(_mutex);
            return _total_count;
        }

        // 获取待推送消息数量
        size_t pending_count()
        {
            std::lock_guard<std::mutex> lock(_mutex);
            return _pendingToPush_list.size();
        }

        // 获取待确认消息数量
        size_t waitcak_count()
        {
            std::lock_guard<std::mutex> lock(_mutex);
            return _waitack_msg_map.size();
        }

        // 获取持久化消息数量
        size_t durable_count()
        {
            std::lock_guard<std::mutex> lock(_mutex);
            return _durable_msg_map.size();
        }

        void clear()
        {
            std::lock_guard<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _pendingToPush_list.clear();
            _durable_msg_map.clear();
            _waitack_msg_map.clear();
            _valid_count = _total_count = 0;
        }

    private:
        // 判断是否需要进行垃圾回收
        bool _cheack_for_gc()
        {
            // 当前总消息数量大于2000，且有效消息数量占总消息数量的50%以下
            //if (_valid_count * 10 / _total_count < 5)     // 临时测试
            if (_total_count > 2000 && (_valid_count * 10 / _total_count < 5))
            {
                return true;
            }
            return false;
        }
        void gc()
        {
            // 1.判断是否需要进行垃圾回收
            if (!_cheack_for_gc())
            {
                return;
            }
            // 2.进行垃圾回收
            std::list<MessagePtr> gc_list = _mapper.gc(); // 对持久化文件进行垃圾回收
            // 3.更新每条消息在文件中对应的offset和length
            for (auto &p_msg : gc_list)
            {
                auto it = _durable_msg_map.find(p_msg->payload().propeties().id());
                if (it == _durable_msg_map.end())
                {
                    // 不该出现的情况，如果出现说明持久化在文件中的消息不在内存中，需要重新加载
                    ILOG("QueueMessage::gc() -> 持久化消息 %s 不在内存中，需要重新加载", p_msg->payload().propeties().id().c_str());
                    _pendingToPush_list.push_back(p_msg);
                    _durable_msg_map.insert(std::make_pair(p_msg->payload().propeties().id(), p_msg));
                    continue;
                }
                it->second->set_offset(p_msg->offset());
                it->second->set_length(p_msg->length());
            }
            // 4.更新统计信息
            size_t old_total_count = _total_count;
            _valid_count = _total_count = gc_list.size();
            // 5.输出日志
            ILOG("QueueMessage::gc() -> 回收了 %ld 个消息", old_total_count - _total_count);
        }

    private:
        std::mutex _mutex;
        std::string _queue_name;
        size_t _valid_count;        // 表示持久化的有效消息数量
        size_t _total_count;        // 表示持久化总消息数量（包括有效消息和已确认<删除>消息）
        MessageMapper _mapper;
        std::list<MessagePtr> _pendingToPush_list; // 待推送消息列表 -> {对应的getable_count()包含持久化和非持久化消息}
        MessageMap _durable_msg_map;               // 永久存储的消息哈希表
        MessageMap _waitack_msg_map;               // 待确认的消息哈希表
    };

    // 消息管理器
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;

    public:
        MessageManager(const std::string &basedir) : _basedir(basedir) {}

        // 将未初始化的队列消息初始化
        void initQueueMessage(const std::string &queue_name)
        {
            QueueMessage::ptr p_qm;
            {
                std::lock_guard<std::mutex> lock(_mutex);
                auto it = _queue_map.find(queue_name);
                if (it != _queue_map.end())
                {
                    ILOG("MessageManager::initQueueMessage() -> 队列 %s 已存在", queue_name.c_str());
                    return; // 队列已存在
                }
                p_qm = std::make_shared<QueueMessage>(_basedir, queue_name);
                _queue_map.insert(std::make_pair(queue_name, p_qm));
            }
            p_qm->recovery(); // 加载持久化消息，此时p_qm必定不为空
        }

        void destroyQueueMessage(const std::string &queue_name)
        {
            QueueMessage::ptr p_qm;
            {
                std::lock_guard<std::mutex> lock(_mutex);
                auto it = _queue_map.find(queue_name);
                if (it == _queue_map.end())
                {
                    ILOG("MessageManager::destroyQueueMessage() -> 不存在该消息管理句柄<%s>", queue_name.c_str());
                    return; // 不存在该队列
                }
                p_qm = it->second;
                _queue_map.erase(it);
            }
            p_qm->clear(); // 加载持久化消息，此时p_qm必定不为空
        }

        bool insert(const std::string &queue_name, const xzt::BasicPropeties *basic_props,
                    const std::string &body, DeliveryMode delivery_mode = DeliveryMode::DURABLE)
        {
            QueueMessage::ptr p_qm;
            {
                std::lock_guard<std::mutex> lock(_mutex);
                auto it = _queue_map.find(queue_name);
                if (it == _queue_map.end())
                {
                    ELOG("MessageManager::insert() -> 不存在该消息管理句柄<%s>", queue_name.c_str());
                    return false; // 不存在该队列
                }
                p_qm = it->second;
            }
            p_qm->insert(basic_props, body, delivery_mode);
            return true;
        }

        MessagePtr front(const std::string &queue_name)
        {
            QueueMessage::ptr p_qm;
            {
                std::lock_guard<std::mutex> lock(_mutex);
                auto it = _queue_map.find(queue_name);
                if (it == _queue_map.end())
                {
                    ELOG("MessageManager::front() -> 不存在该消息管理句柄<%s>", queue_name.c_str());
                    return MessagePtr(); // 不存在该队列
                }
                p_qm = it->second;
            }

            auto p_msg = p_qm->front();
            return p_msg;
        }

        bool ack(const std::string &queue_name, const std::string &msg_id)
        {
            bool ret = false;
            QueueMessage::ptr p_qm;
            {
                std::lock_guard<std::mutex> lock(_mutex);
                auto it = _queue_map.find(queue_name);
                if (it == _queue_map.end())
                {
                    ELOG("MessageManager::ack() -> 不存在该消息管理句柄<%s>", queue_name.c_str());
                    return ret; // 不存在该队列
                }
                p_qm = it->second;
                ret = p_qm->remove(msg_id);
            }
            return ret;
        }

        void clear()
        {
            std::lock_guard<std::mutex> lock(_mutex);
            for (auto &item : _queue_map)
            {
                item.second->clear();
            }
            _queue_map.clear();
        }

        size_t getable_count(const std::string &queue_name)
        {
            QueueMessage::ptr p_qm;
            {
                std::lock_guard<std::mutex> lock(_mutex);
                auto it = _queue_map.find(queue_name);
                if (it == _queue_map.end())
                {
                    ELOG("MessageManager::getable_count() -> 不存在该消息管理句柄<%s>", queue_name.c_str());
                    return 0; // 不存在该队列
                }
                p_qm = it->second;
            }
            return p_qm->getable_count();
        }

        size_t total_count(const std::string &queue_name)
        {
            QueueMessage::ptr p_qm;
            {
                std::lock_guard<std::mutex> lock(_mutex);
                auto it = _queue_map.find(queue_name);
                if (it == _queue_map.end())
                {
                    ELOG("MessageManager::total_count() -> 不存在该消息管理句柄<%s>", queue_name.c_str());
                    return 0; // 不存在该队列
                }
                p_qm = it->second;
            }
            return p_qm->total_count();
        }

        size_t waitcak_count(const std::string &queue_name)
        {
            QueueMessage::ptr p_qm;
            {
                std::lock_guard<std::mutex> lock(_mutex);
                auto it = _queue_map.find(queue_name);
                if (it == _queue_map.end())
                {
                    ELOG("MessageManager::waitcak_count() -> 不存在该消息管理句柄<%s>", queue_name.c_str());
                    return 0; // 不存在该队列
                }
                p_qm = it->second;
            }
            return p_qm->waitcak_count();
        }

        size_t durable_count(const std::string &queue_name)
        {
            QueueMessage::ptr p_qm;
            {
                std::lock_guard<std::mutex> lock(_mutex);
                auto it = _queue_map.find(queue_name);
                if (it == _queue_map.end())
                {
                    ELOG("MessageManager::durable_count() -> 不存在该消息管理句柄<%s>", queue_name.c_str());
                    return 0; // 不存在该队列
                }
                p_qm = it->second;
            }
            return p_qm->durable_count();
        }

    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_map;
    };
}
